package p.ithorns.framework.logger.appender;

import com.google.common.util.concurrent.ThreadFactoryBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;


import java.io.BufferedReader;
import java.io.IOException;
import java.io.PipedReader;
import java.io.PipedWriter;
import java.util.concurrent.ThreadFactory;
import java.util.function.Consumer;

/**
 * LogPipeline
 *
 * 日志管道
 *
 * @author 2021/10/23 16:41 by lht
 */
public class LogPipeline {

    private final static Logger log = LoggerFactory.getLogger(LogPipeline.class);

    private final static ThreadFactory THREAD_FACTORY = new ThreadFactoryBuilder()
            .setDaemon(false)
            .setNameFormat("log-pipe-%d")
            .build();

    private final PipedWriter writer;
    private final PipedReader reader;
    private Thread thread;


    public LogPipeline() throws IOException {
        this.reader = new PipedReader();
        this.writer = new PipedWriter(reader);
    }


    public PipedWriter getWriter() {
        return writer;
    }


    /**
     * 连接消费者
     *
     * 开启一个新的线程处理日志数据回调，
     * 当管道有日志内容时，交由消费者具体实现来处理
     *
     * @param consumer -消费者
     */
    public void connect(final Consumer<String> consumer) {
        this.thread = THREAD_FACTORY.newThread(() -> {
            this.apply(consumer);
        });
        this.thread.start();
    }


    /**
     * 消费者回调
     *
     * @param consumer -消费者
     */
    private void apply(final Consumer<String> consumer) {
        String line;
        try (BufferedReader br = new BufferedReader(reader)) {
            while ((line = br.readLine()) != null) {
                consumer.accept(line);
            }
        } catch (IOException e) {
            log.warn("==> Pipeline read error: {}", e.getMessage());
        }
    }

    /**
     * 关闭
     */
    public void close() {
        try {
            this.writer.close();
            this.reader.close();
            this.thread.interrupt();
            this.thread.join();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

}